package com.bbx.flink.demo.data_stream_state.functions;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;


/**
 * keyed state  -valueState
 * 统计输入元素 的个数（ keyBy(...)  后计数）
 */
@Slf4j
public class CountWindowAverage extends  RichFlatMapFunction<Integer,Integer>  {
    private static final long serialVersionUID = -7065486913783002819L;

    private ValueState<Integer> valueState;

    @Override
    public void open(Configuration parameters) throws Exception {
        valueState= getRuntimeContext()
                .getState(new ValueStateDescriptor<Integer>("",Integer.class));
    }

    @Override
    public void flatMap(Integer value, Collector<Integer> out) throws Exception {
            Integer valueStateValue = valueState.value();
        if (valueStateValue == null) {
            valueStateValue = 0;
        }
        valueStateValue+=1;
        valueState.update(valueStateValue);
        out.collect(valueStateValue);
    }


}
